package com.qch.rocket.demo.producer;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;


/**
 * @Author qch
 * @Date 2021/6/7
 */
public class MyProducer {
    public static void main(String[] args) throws UnsupportedEncodingException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
        //在实例化生产者的同时，指定了生产组名称
        DefaultMQProducer producer = new DefaultMQProducer("myproducer_grp_06");
        //指定namerServer的地址
        producer.setNamesrvAddr("192.168.142.134:9876");
        //对生产者进行初始化
        producer.start();
        for (int i = 0; i <100 ; i++) {
            Message message = new Message(
                    "tp_demo_06",
                    "tag-"+(i%3),
                    ("hello my mes"+i).getBytes(RemotingHelper.DEFAULT_CHARSET));


            //发送消息
            //批量发送消息
            producer.send(message, new SendCallback() {
                public void onSuccess(SendResult sendResult) {
                    System.out.println(sendResult.getSendStatus());
                }

                public void onException(Throwable throwable) {
                    System.out.println(throwable.getMessage());
                }
            });

        }
        Thread.sleep(30000);
        //关闭生产者
        producer.shutdown();

    }
}
